import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaToMySQL {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.88.131:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "huihui");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 根据需要调整

        Consumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList("ETC"));

        ObjectMapper objectMapper = new ObjectMapper();

        try {
            Connection connection = DriverManager.getConnection("jdbc:mysql://192.168.88.131:3306/huihui?useSSL=false", "root", "12345678");

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                records.forEach(record -> {
                    String jsonString = record.value();

                    try {
                        JsonNode jsonNode = objectMapper.readTree(jsonString);

                        // 从 JSON 中提取值
                        String XH = jsonNode.get("XH").asText();
                        String CP = jsonNode.get("CP").asText();
                        String CX = jsonNode.get("CX").asText();
                        String RKSJ = jsonNode.get("RKSJ").asText();
                        String SFZRKMC = jsonNode.get("SFZRKMC").asText();
                        String CKSJ = jsonNode.get("CKSJ").asText();
                        String BZ = jsonNode.get("BZ").asText();

                        // 插入到 MySQL 数据库
                        String query = "INSERT INTO etc (XH, CP, CX, RKSJ, SFZRKMC, CKSJ, BZ) VALUES (?, ?, ?, ?, ?, ?, ?)";
                        try (PreparedStatement preparedStatement = connection.prepareStatement(query)) {
                            preparedStatement.setString(1, XH);
                            preparedStatement.setString(2, CP);
                            preparedStatement.setString(3, CX);
                            preparedStatement.setString(4, RKSJ);
                            preparedStatement.setString(5, SFZRKMC);
                            preparedStatement.setString(6, CKSJ);
                            preparedStatement.setString(7, BZ);

                            preparedStatement.executeUpdate();
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}
